package main

import (
	"strings"
	"sync"
	"time"

	"../prometheus"
	cluster "github.com/bsm/sarama-cluster"
	"github.com/kpango/glg"
)

type KafkaHandler struct {
	consumer *cluster.Consumer
	closed   chan interface{}
	matrix   *Matrix
}

type Matrix struct {
	ms     *prometheus.MatrixSet
	mu     *sync.RWMutex
	count  map[string]int64
	delay  map[string]int64
	offset map[string]int64
}

func NewKafkaHandler() (KafkaHandler, error) {
	handler := KafkaHandler{}
	handler.closed = make(chan interface{})

	handler.matrix = &Matrix{
		ms:     prometheus.NewMatrixSet("crmlog-consumer"),
		mu:     new(sync.RWMutex),
		count:  make(map[string]int64),
		delay:  make(map[string]int64),
		offset: make(map[string]int64),
	}
	consumerConfig := cluster.NewConfig()
	consumerConfig.Consumer.Return.Errors = true
	brokers := strings.Split(*KUrl, ",")
	topics := strings.Split(*KTopic, ",")
	consumer, err := cluster.NewConsumer(brokers, "crmlog-consumer-group", topics, consumerConfig)
	if err != nil {
		return handler, err
	}
	handler.consumer = consumer
	return handler, nil
}

func (kh *KafkaHandler) Close() error {
	go func() {
		close(kh.closed)
	}()
	if err := kh.consumer.Close(); err != nil {
		return err
	}
	return nil
}

func (kh *KafkaHandler) Consume() {
	tick := time.Tick(time.Minute * time.Duration(*Interval))
	for {
		select {
		case <-tick:
			time.Sleep(time.Minute * 20)
		case err := <-kh.consumer.Errors():
			if err != nil {
				glg.Errorf("kafkaConsume.Consume: kafka consumer error: %v", err)
			}
		case msg := <-kh.consumer.Messages():
			t := time.Now().UnixNano() / 1000 / 1000
			err := WriteToEverylog(string(msg.Key), string(msg.Value))
			delta := time.Now().UnixNano()/1000/1000 - t
			if err != nil {
				kh.matrix.Fail(delta)
				glg.Errorf("kafkaConsume.Consume: Failed to write to Everylog: %v", err)
				glg.Errorf("kafkaConsume.Consume: 插入数据失败，数据为: offset:%v, tableInfo:%s, data:%s", msg.Offset, string(msg.Key), string(msg.Value))
			} else {
				kh.matrix.Success(msg.Offset, delta)
				glg.Infof("kafkaConsume.Consume: 插入数据成功，数据为: offset:%v, tableInfo:%s, data:%s", msg.Offset, string(msg.Key), string(msg.Value))
				kh.consumer.MarkOffset(msg, "")
			}
		case <-kh.closed:
			return
		}
	}
}

//++
func (m *Matrix) Success(offset, delta int64) {
	m.mu.Lock()
	m.count["count_total"]++
	m.count["count_success"]++
	m.delay["cost_total"] = m.delay["cost_total"] + delta
	m.delay["cost_success"] = m.delay["cost_success"] + delta
	m.offset["offset"] = offset
	m.ms.UpdateMatrix(m.count, "", "message_sent_count")
	m.ms.UpdateMatrix(m.delay, "ms", "message_sent_delay")
	m.ms.UpdateMatrix(m.offset, "", "message_offset")
	m.mu.Unlock()
}

func (m *Matrix) Fail(delta int64) {
	m.mu.Lock()
	m.count["count_total"]++
	m.count["count_fail"]++
	m.delay["cost_total"] = m.delay["cost_total"] + delta
	m.delay["cost_fail"] = m.delay["cost_fail"] + delta
	m.ms.UpdateMatrix(m.count, "", "message_sent_count")
	m.ms.UpdateMatrix(m.delay, "ms", "message_sent_delay")
	m.mu.Unlock()
}

